package com.bgds.sys.data.iXCom29d.gateway;

import com.alibaba.fastjson.JSON;
import com.bgds.sys.data.iXCom29d.gateway.impl.*;
import com.bgds.sys.data.iXCom29d.gateway.vo.IXCom29dCustomParam;
import com.cisdi.data.common.exception.BusinessException;
import com.cisdi.data.common.utils.Strings;

import com.cisdi.data.sdk.consts.ServiceName;
import com.cisdi.data.sdk.enums.GatewayState;
import com.cisdi.data.sdk.gateway.base.SocketGatewayBase;
import com.cisdi.data.sdk.gateway.netty.IoSession;
import com.cisdi.data.sdk.gateway.netty.SessionFactory;
import com.cisdi.data.sdk.gateway.netty.TcpIoService;
import com.cisdi.data.sdk.gateway.netty.impl.DefaultTcpIoService;
import com.cisdi.data.sdk.param.ListenSocketParam;
import com.cisdi.data.sdk.procotol.message.SocketReturnMessage;
import com.cisdi.data.sdk.service.RouteService;
import com.cisdi.data.sdk.vo.DeviceVo;
import com.cisdi.data.sdk.vo.ExeResultVo;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author
 */
public class IXCom29dSocketGateway extends SocketGatewayBase {
	
	private static Logger logger = LoggerFactory.getLogger(IXCom29dSocketGateway.class);
	
	TcpIoService ioService = null;
	SessionFactory sessionFactory = null;
	AtomicBoolean shouldRun = null;
	private String deviceId;
	
	Charset charset = Charset.forName("GBK");
	
	@Override
	public ExeResultVo sendReturnMessage(SocketReturnMessage returnMsg) {
		throw new BusinessException("宝信29d网关不支持下发");
	}

	@Override
	public void start() {
		if(state == GatewayState.RUNNING) {
			return;
		}
		
		ListenSocketParam socketParam = JSON.parseObject(getInstanceVo().getParameter(), ListenSocketParam.class);
		String parameter = socketParam.getOtherParameter();
		
		if(parameter != null) {
			IXCom29dCustomParam customParam = JSON.parseObject(parameter, IXCom29dCustomParam.class);
			
			if(customParam != null && Strings.isNotEmpty(customParam.getEncode())) {
				charset = Charset.forName(customParam.getEncode());
			}
		}
		
		RouteService routeService = (RouteService)serviceProvider.getByName(ServiceName.Route);
		List<DeviceVo> deviceVos = routeService.findByRunId(instanceVo.getRunId());
		
		if(deviceVos.size() > 1) {
			throw new BusinessException("网关Id:" +instanceVo.getRunId() + "为宝信29d网关，只允许关联一个设备Id");
		}
		
		deviceId = deviceVos.size() > 0 ? deviceVos.get(0).getDeviceId() : null;
		
		if(StringUtils.isEmpty(deviceId)) {
			throw new BusinessException("runId=" + getInstanceVo().getRunId() + "未正确配置网关和设备关联");
		}
		
		logger.info("宝信29d网关:{}读取启动配置参数:{}",instanceVo.getRunId(), getInstanceVo().getParameter());
		
		DefaultTcpIoService defaultIoService = new DefaultTcpIoService();
		SessionFactory factory = new BaoXingSessionFactory(deviceId, charset);
		factory.init(serviceProvider, this);
		
		IXCom29dChannelInitializer<BaoXingVo> channelInitializer =
				new IXCom29dChannelInitializer<BaoXingVo>(factory, charset);
		
		defaultIoService.init(instanceVo, factory, channelInitializer);
		
		ioService = defaultIoService;
		sessionFactory = factory;
		
		boolean open = ioService.open();
		
		
		if(open == true) {
			state = GatewayState.RUNNING;
			logger.info("宝信29d网关:{}启动成功 参数:{}", instanceVo.getRunId(), getInstanceVo().getParameter());
			
			shouldRun = new AtomicBoolean(true);
			
			Integer keepAlive = socketParam.getKeepAlive();
			
			if(keepAlive == null || keepAlive < 120) {
				keepAlive = 120;
			}else if (keepAlive > 3600) {
				keepAlive = 3600;
			}
			
			AliveCheckTask task = new AliveCheckTask(keepAlive, shouldRun, factory);

			ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
					.setNameFormat("宝信29d-alive-check-thread-%d").build();
			ExecutorService singleThreadPool = new ThreadPoolExecutor(1, 10,
					0L, TimeUnit.MILLISECONDS,
					new LinkedBlockingQueue<Runnable>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

			singleThreadPool.execute(task);
			singleThreadPool.shutdown();
		}else {
			logger.info("宝信29d网关:{}启动失败 参数:{}", instanceVo.getRunId(), getInstanceVo().getParameter());
		}
	}

	@Override
	public void shutdown() {
		if(state == GatewayState.CLOSED) {
			return;
		}
		
		if(sessionFactory != null) {
        	for (IoSession ioSession : sessionFactory.getSessions()) {
        		BaoXingSession baoXin29dIoSession = (BaoXingSession)ioSession;
				baoXin29dIoSession.close();
        	}		       	
		}
		
		boolean close = ioService.close();
		if(close == true) {
			sessionFactory = null;
			state = GatewayState.CLOSED;
			logger.info("宝信29d网关:{}关闭成功", instanceVo.getRunId());
		}else {
			logger.info("宝信29d网关:{}关闭失败", instanceVo.getRunId());
		}
	}
	
	@Override
    public Set<String> getActiveDeviceIds(){
		
		Set<String> set = Sets.newConcurrentHashSet();
		
		if(StringUtils.isNotEmpty(deviceId)) {
			set.add(deviceId);
		}
		
		return set;
	}
}
